/*
 *  Arrow
 *
 *  Created by David Giovannini on 1/5/07.
 *  Copyright 2007 Software by Jove. All rights reserved.
 *
 */
 
#ifndef SBJ_Thread_TaskDispatcher_h
#define SBJ_Thread_TaskDispatcher_h

#include "SBJ/Interfaces/IExceptionObserver.h"

#include <boost/function.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/thread/condition.hpp>

#include <list>
#include <vector>

namespace SBJ
{

class PolyTask;

/*
Class: TaskDispatcher

TaskDispatcher is a set of threads that executes a variable number of 
tasks. On construction the TaskDispatcher creates the specified number of threads.
All the threads go into a wait state until tasks are added. The tasks are
handled first come first serve. The join_all_tasks forces the calling thread
to wait until all known tasks are completed.

To create a polymorphic TaskDispatcher
Declare: PolyTaskDispatcher myDispatcher;
Any polymorphic tasks must be derived from PolyTask
To add a polymorphic task to the dispatcher use the form
myDispatcher.addPolyTask( new MyTask );

TODO:
Provide interface and implementation for optional thread affinity.
*/

template < typename Task >
class TaskDispatcher
{
public:
	/// Parameterized constructor, requires number of threads
	explicit TaskDispatcher( int numThreads, bool clearUnusedOnDestroy = false );

	/// Destructor
	~TaskDispatcher( void );

	/// Enqueues a copy of the task
	void addTask( Task& task );

	/// Enqueues a PolyTaskRef that owns the incoming task
	/// The TaskDispatcher must be instantiated as TaskDispatcher< PolyTaskRef > to use this method
	void addPolyTask( PolyTask* task );

	/// Dequeues all tasks not being executed
	void clear_remaining_tasks( void );

	/// Wait until all queued tasks are complete
	void join_all_tasks( void );

private:
	bool invariant( void ) const;

	void threadWork( int threadIndex );

	IExceptionObserver::Ptr m_excObserver;

	std::vector< boost::thread* > m_threads;

	boost::mutex m_taskGuard;
	boost::condition m_stop;
	boost::condition m_start;

	typedef std::list< Task > Tasks;
	Tasks m_tasks;
	int m_pending;
	bool m_clearUnusedOnDestroy;
};

class PolyTask
{
public:
	virtual ~PolyTask( void ) {}
	virtual void operator () ( void ) = 0;
};

class PolyTaskRef
{
public:
	PolyTaskRef( PolyTask* task = 0 ) : m_task( task ) {}
	PolyTaskRef( const PolyTaskRef& rhs ) : m_task( 0 ) { std::swap( m_task, rhs.m_task ); }
	~PolyTaskRef( void ) { delete m_task; }
	PolyTaskRef& operator = ( const PolyTaskRef& rhs ) { std::swap( m_task, rhs.m_task ); return *this; }
	void operator () ( void )
	{
		if ( m_task )
		{
			PolyTask* temp = m_task;
			m_task = 0;
			(*temp)();
			delete temp;
		}
	}
private:
	mutable PolyTask* m_task;
};

typedef TaskDispatcher< PolyTaskRef > PolyTaskDispatcher;

} // SBJ

#include "SBJ/Thread/TaskDispatcher.inl"

#endif // SBJ_Thread_TaskDispatcher_